package com.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaProducerTest implements Runnable {

    private final KafkaProducer<String, String> producer;
    private final String topic;

    public KafkaProducerTest(String topicName) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.11.31:9092,192.168.11.32:9092,192.168.11.33:9092");
        props.put("acks", "all");
        /*acks指定了必须有多少个分区副本接收到了消息，生产者才会认为消息是发送成功的。
        acks=0，生产者成功写入消息之前不会等待来自任何服务器的响应，这种配置，提高吞吐量，但是消息存在丢失风险。
        acks=1，只要集群的leader（master）收到了消息，生产者将会受到发送成功的一个响应，如果消息无撞到达首领节点（比如首领节点崩愤，新的首领还没有被选举出来），生产者会收到一个错误响应，为了避免数据丢失，生产者会重发消息。不过，如果一个没有收到消息的节点成为新首领，消息还是会丢失。这个时候的吞吐量取决于使用的是
        同步发送还是异步发送。如果让发送客户端等待服务器的响应（通过调用Futu re 对象的get（）方法，显然会增加延迟（在网络上传输一个来回的延迟）。如果客户端使用回调，延迟问题就可以得到缓解，不过吞吐量还是会受发送中消息数量的限制（比如，生产者在收到服务器响应之前可以发送多少个消息）。
        acks=all，所有参与复制的节点全部收到消息的时候，生产者才会收到来自服务器的一个响应，这种模式最安全，但是吞吐量受限制，它可以保证不止一个服务器收到消息，就算某台服务器奔溃，那么整个集群还是会正产运转。*/

        props.put("retries", 0);
        /*生产者从服务器收到的错误消息有可能是临时的，当生产者收到服务器发来的错误消息，会启动重试机制，
        当充实了n（设置的值）次，还是收到错误消息，那么将会返回错误。
        生产者会在每次重试之间间隔100ms，不过可以通过retry.backoff.ms改变这个间隔。*/

        props.put("batch.size", 16384);
        /*当多个消息发往 同一个分区，生产者会将他们放进同一个批次，
        该参数指定了一个批次可以使用的内存大小，按照字节数进行计算，不是消息个数，
        当批次被填满，批次里面所有得消息将会被发送，半满的批次，甚至只包含一个消息也可能会被发送，
        所以即使把批次设置的很大，也不会造成延迟，只是占用的内存打了一些而已。
        但是设置的太小，那么生产者将会频繁的发送小，增加一些额外的开销。*/
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());
        this.producer = new KafkaProducer<String, String>(props);
        this.topic = topicName;
    }

    @Override
    public void run() {
        int messageNo = 1;
        try {
            for (; ; ) {
                String messageStr = "你好，这是第" + messageNo + "条数据";
                producer.send(new ProducerRecord<String, String>(topic, "Message", messageStr));
                //生产了100条就打印
                if (messageNo % 100 == 0) {
                    System.out.println("发送的信息:" + messageStr);
                }
                //生产1000条就退出
                if (messageNo % 1000 == 0) {
                    System.out.println("成功发送了" + messageNo + "条");
                    break;
                }
                messageNo++;
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }

      /*在kafka 的消息发送端的 ProducerConfig.ACKS_CONFIG(acks),"-1"   这个设置的参数有下面的几个作用
     当参数是0 的时候  表示消息发送给broker 以后，不需要进行确认（性能较高，但是会出现数据丢失的情况）
     当参数是1 的时候  表示只需要获得 kafka 集群中的 leader 节点的确认即可返回  (leader、follower)
     all(-1) 需要ISR 中的所有 Replica 去进行确认(需要集群当中所有的节点进行确认)，最安全的，但是也可能会出现数据丢失的情况*/


    public static void main(String args[]) {
        KafkaProducerTest test = new KafkaProducerTest("KAFKA_TEST");
        Thread thread = new Thread(test);
        thread.start();
    }
}